package org.carl.base.driving.listener;

import cn.hutool.core.util.IdUtil;
import cn.hutool.json.JSONUtil;
import com.alibaba.fastjson.JSONObject;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.carl.base.driving.core.ReceiveExecute;
import org.carl.base.driving.entity.ReceiveMessage;
import org.carl.base.driving.entity.SendMessage;
import org.carl.base.driving.enums.ReceiveHandleStateEnum;
import org.carl.base.driving.service.receive.ReceiveHandle;
import org.carl.base.driving.service.receive.ReceiveMessageService;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;
import org.springframework.util.StringUtils;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.Date;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * 消息接收监听器
 */
@Slf4j
@Service
@ConditionalOnBean(value ={ReceiveExecute.class})
public class ReceiveListener implements ChannelAwareMessageListener {
    private static final ExecutorService MQ_THREAD_POOL= Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());


    @Autowired
    private ReceiveExecute receiveExecute;

    @Autowired
    private ReceiveMessageService receiveMessageService;

    @Override
    public void onMessage(Message message, Channel channel) throws Exception {
        log.debug("消息驱动接收消息,数据{}", message.getBody());
        if (message == null || StringUtils.isEmpty(message.getBody())) {
            return;
        }
        // 获取发送数据
        String json = new String(message.getBody(), StandardCharsets.UTF_8);
        log.info("消息驱动接收消息,数据{}", json);
        SendMessage sendMessage = null;
        if(JSONUtil.isJson(json)){
            sendMessage = JSONObject.parseObject(json, SendMessage.class);
        }
        if (sendMessage == null) {
            // 拒绝并且丢弃
            channel.basicReject(message.getMessageProperties().getDeliveryTag(), false);
            return;
        }
        // 当前服务没有订阅则直接丢弃
        ReceiveHandle receiveHandle = receiveExecute.getReceiveHandle(sendMessage.getEventName());
        if (receiveHandle == null) {
            log.warn("消息驱动未订阅该消息id:{},消息内容:{}", sendMessage.getSendId(),message);
            try {
                // 拒绝并且丢弃
                channel.basicReject(message.getMessageProperties().getDeliveryTag(), false);
            } catch (IOException e) {
                e.printStackTrace();
            }
            return;
        }
        // rabbitMq是通过轮询消费者方式投递消息，不会出现多个消费者从一个队列消费一个消息情况，这里可以不用加分布式锁
        List<ReceiveMessage> messageList = receiveMessageService.findBySendId(sendMessage.getSendId());
        if (!CollectionUtils.isEmpty(messageList)) {
            log.warn("消息驱动接收消息:{}", messageList);
            try {
                // 拒绝并且丢弃
                channel.basicReject(message.getMessageProperties().getDeliveryTag(), false);
            } catch (IOException e) {
                e.printStackTrace();
            }
            return;
        }
        // 预处理消息
        ReceiveMessage receiveMessage = new ReceiveMessage();
        receiveMessage.setReceiveId(IdUtil.randomUUID());
        receiveMessage.setSendId(sendMessage.getSendId());
        receiveMessage.setHandleName(sendMessage.getEventName());
        receiveMessage.setData(sendMessage.getData());
        receiveMessage.setGmtReceive(new Date());
        receiveMessage.setHandleState(ReceiveHandleStateEnum.INIT.getCode());
        receiveMessageService.insert(receiveMessage);
        log.debug("消息驱动接收消息 - 消息写入DB成功{}", receiveMessage);
        // 消息确认
        try {
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        } catch (IOException e) {
            log.error("ACK消息失败", e);
        }
        // 异步执行业务，忽略执行结果
        CompletableFuture.runAsync(() -> receiveExecute.execute(receiveMessage.getReceiveId(), receiveMessage.getHandleName()),MQ_THREAD_POOL);
    }
}
